/*
 * KeyValuePageVoidKeyFixedValueImpl_.h
 *
 *  Created on: Aug 1, 2011
 *      Author: yong
 */

#ifndef KEYVALUEPAGEVOIDKEYFIXEDVALUEIMPL__H_
#define KEYVALUEPAGEVOIDKEYFIXEDVALUEIMPL__H_
#include "../Page.h"
#include "../MemPageMaster.h"
#include "../KeyValue.h"
#include "../Typedef.h"
#include "../Mapper.h"
#include "../traits/is_fixed_size.h"
#include <iostream>
#include <boost/type_traits.hpp>
#include <boost/utility.hpp>
#include <iostream>
#include <cstring>
#include <cstdio>
#include <vector>
#include <string>
#include <algorithm>
using namespace std;
namespace distrim
{

template<typename V, typename P>
class KeyValuePage<void, V, P, true, true> : public Page<void, V>
{
public:
	KeyValuePage(size_t pageID, void *ptrPageBeg, const string &fileName,
			Mapper<void, V, P> *ptrConsumer, size_t pairsCount,
			size_t valueElemCount, size_t maxPairCount) :
		Page<void, V> (pageID, ptrPageBeg, fileName),
				m_ptrConsumer(ptrConsumer), m_pairsCount(pairsCount),
				m_maxPairCount(maxPairCount), m_valueElemCount(valueElemCount),
				m_keyValue(valueElemCount)
	{
	}
	// Default constructor, for new[] operator to create array of KeyValuePage records.
	KeyValuePage() :
		Page<void, V> (), m_ptrConsumer(NULL), m_pairsCount(0), m_maxPairCount(
				0), m_valueElemCount(0)
	{

	}
	virtual ~KeyValuePage()
	{
	}

private:
	/**
	 *
	 */
	Mapper<void, V, P> *m_ptrConsumer;
	/**
	 * Number of key-value pairs within this page.
	 */
	size_t m_pairsCount;
	/**
	 * Maximum count of key-pairs.
	 */
	size_t m_maxPairCount;
	/**
	 * Counts of value elements.
	 */
	size_t m_valueElemCount;
	/**
	 *
	 */
	KeyValue<void, V> m_keyValue;

public:

	/**
	 *
	 */
	inline void SetConsumer(Mapper<void, V, P> *ptrConsumer)
	{
		m_ptrConsumer = ptrConsumer;
	}
	/**
	 *
	 */
	inline Mapper<void, V, P> *GetConsumer() const
	{
		return m_ptrConsumer;
	}
	/**
	 *	Set maximum pairs this page can accommadate.
	 */
	inline void SetMaxPairCount(size_t maxPairCount)
	{
		m_maxPairCount = maxPairCount;
	}
	/**
	 * Get maximum pairs this page can accommadate.
	 */
	inline const size_t GetMaxPairCount() const
	{
		return m_maxPairCount;
	}
	/**
	 * Set count of value elements.
	 */
	inline void SetValueElemCount(size_t valueElemCount)
	{
		m_valueElemCount = valueElemCount;
		m_keyValue.SetValueElemCount(valueElemCount);
	}
	/**
	 * Return number of value elements.
	 */
	inline const size_t GetValueElemCount() const
	{
		return m_valueElemCount;
	}
	/**
	 * Get number of pairs within this page.
	 */
	inline size_t GetPairsCount() const
	{
		return m_pairsCount;
	}
	/**
	 * @param  index
	 * @return A pair of pointers pointing the value range in the page.
	 * @brief  Fast obtain key-value pair without check.
	 */
	inline const KeyValue<void, V> &operator[](size_t index)
	{
		V *valueBegin = (V*) Page<void, V>::m_ptrPageBeg + index
				* m_valueElemCount;
		m_keyValue.SetValueBegin(valueBegin);
		return m_keyValue;
	}
	/**
	 * @param  index Index of value arrays to obtain.
	 * @return  A pair of pointers pointing the value range in the page.
	 * @brief  Obtain key-value pair with check.
	 */
	inline KeyValue<void, V>& At(size_t index)
	{
		if (index < 0 || index > m_pairsCount - 1)
		{
			throw std::invalid_argument("index out of range");
		}
		V *valueBegin = (V*) Page<void, V>::m_ptrPageBeg + index
				* m_valueElemCount;
		m_keyValue.SetValueBegin(valueBegin);
		return m_keyValue;
	}
	/**
	 * @param  index Index of value arrays to obtain.
	 * @return  A pair of pointers pointing the value range in the page.
	 * @brief  Obtain key-value pair with check.
	 */
	inline const KeyValue<void, V>& At(size_t index) const
	{
		if (index < 0 || index > m_pairsCount - 1)
		{
			throw std::invalid_argument("index out of range");
		}
		V *valueBegin = (V*) Page<void, V>::m_ptrPageBeg + index
				* m_valueElemCount;
		m_keyValue.SetValueBegin(valueBegin);
		return m_keyValue;
	}
	/**
	 * @param  valuesFrom
	 * @param  valueElemCount
	 * @throw  std::logic_error
	 * @brief  Copy values from an array and add to memory page.
	 */
	inline void AddValueArray(const V* valuesFrom, size_t valueElemCount)
			throw (std::logic_error)
	{
		if (m_pairsCount + 1 > m_maxPairCount)
		{
			throw std::logic_error("out of maximum page capacity");
		}
		if (valueElemCount != m_valueElemCount)
		{
			throw std::invalid_argument("incompatible number of value elements");
		}
		V* valuesTo = (V*) Page<void, V>::m_ptrPageBeg + m_pairsCount
				* m_valueElemCount;
		//		memcpy(valuesTo, valuesFrom, sizeof(V) * m_valueElemCount);
		std::copy(valuesFrom, valuesFrom + m_valueElemCount, valuesTo);
		++m_pairsCount;
	}

	inline void Reset()
	{
		m_pairsCount = 0;
	}
	/**
	 *	@brief  Define how key-value pairs within this page be written to disk in binary mode
	 *	as a continuous chunk.
	 */
	virtual void WriteToDisk() throw (std::logic_error, std::runtime_error)
	{
		if (Page<void, V>::m_fileName == string(""))
		{
			throw std::logic_error("empty file name to write");
		}
		FILE * pf;
		pf = fopen(Page<void, V>::m_fileName.c_str(), "wb");
		if (pf == NULL)
		{
			throw std::runtime_error("error opening file to write");
		}
		fwrite(&m_pairsCount, 1, sizeof(size_t), pf);
		// Write content of key-values.
		fwrite(Page<void, V>::m_ptrPageBeg, sizeof(char), sizeof(V)
				* m_valueElemCount * m_pairsCount, pf);
		fclose(pf);
	}

	/**
	 * @brief  Define how key-value pairs be restored from a continuous chunk in disk in binary mode.
	 */
	virtual void ReadFromDisk() throw (std::logic_error, std::runtime_error)
	{
		if (Page<void, V>::m_fileName == string(""))
		{
			throw std::logic_error("no file name to read");
		}
		FILE * pf;
		pf = fopen(Page<void, V>::m_fileName.c_str(), "rb");
		if (pf == NULL)
		{
			throw std::runtime_error("error opening file to read");
		}
		fread(&m_pairsCount, sizeof(size_t), 1, pf);
		fread(Page<void, V>::m_ptrPageBeg, sizeof(char), sizeof(V)
				* m_valueElemCount * m_pairsCount, pf);
		fclose(pf);
	}

	inline void Clear()
	{
		if (Page<void, V>::m_fileName.size() != 0)
		{
			::remove(Page<void, V>::m_fileName.c_str());
		}
	}

	virtual void Produce()
	{
		ReadFromDisk();
	}

	virtual void Consume()
	{
		m_ptrConsumer->SetKeyValuePage(this);
		mpi::timer tmr;
		m_ptrConsumer->map();
	}
};
}

#endif /* KEYVALUEPAGEVOIDKEYFIXEDVALUEIMPL__H_ */
